package com.hzmg.akkacloud.worker.common.utils;

import akka.actor.AbstractActor.ActorContext;
import akka.actor.ActorRef;
import akka.actor.ActorSelection;
import akka.cluster.routing.ClusterRouterGroup;
import akka.cluster.routing.ClusterRouterGroupSettings;
import akka.pattern.Patterns;
import akka.routing.BroadcastGroup;
import akka.routing.ConsistentHashingGroup;
import akka.routing.RandomGroup;
import akka.routing.RoundRobinGroup;
import com.google.common.collect.Lists;
import com.hzmg.akkacloud.common.RemoteConstant;
import com.hzmg.akkacloud.common.exception.AkkaCloudException;
import com.hzmg.akkacloud.common.response.AskResponse;
import com.hzmg.akkacloud.common.serialize.SerializerUtils;
import com.hzmg.akkacloud.common.utils.Meaningless;
import com.hzmg.akkacloud.common.utils.SupplierPlus;
import lombok.extern.slf4j.Slf4j;
import org.springframework.util.StringUtils;

import java.time.Duration;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.TimeUnit;
import java.util.function.Supplier;

/**
 * AKKA 工具类
 *
 * @author zbw
 * @since 2022/7/7
 */
@Slf4j
public class AkkaUtils {

    /**
     * akka://<actor system>@<hostname>:<port>/<actor path>
     */
    private static final String AKKA_NODE_PATH = "akka://%s@%s/user/%s";

    public static String getAkkaWorkerPath(String address, String actorName) {
        return String.format(AKKA_NODE_PATH, RemoteConstant.WORKER_ACTOR_SYSTEM_NAME, address, actorName);
    }

    public static String getAkkaWorkerPathWithSystemName(String address, String actorSystemName, String actorName) {
        return String.format(AKKA_NODE_PATH, actorSystemName, address, actorName);
    }

    /**
     * 获取注册中心path
     */
    public static String getServerActorPath(String serverAddress) {
        if (StringUtils.isEmpty(serverAddress)) {
            return null;
        }
        return String.format(AKKA_NODE_PATH, RemoteConstant.SERVER_ACTOR_SYSTEM_NAME, serverAddress, RemoteConstant.SERVER_ACTOR_NAME);
    }

    /**
     * 可靠传输
     *
     * @param remote 远程 AKKA 节点
     * @param msg    需要传输的对象
     * @return true: 对方接收成功 / false: 对方接收失败（可能传输成功但对方处理失败，需要协同处理 AskResponse 返回值）
     */
    public static boolean reliableTransmit(ActorSelection remote, Object msg) {
        try {
            return easyAsk(remote, msg).isSuccess();
        } catch (Exception e) {
            log.warn("[RemoteTransmitter] transmit {} failed", msg, e);
        }
        return false;
    }

    public static AskResponse easyAsk(ActorSelection remote, Object msg) {
        try {
            CompletionStage<Object> ask = Patterns.ask(remote, msg, Duration.ofMillis(RemoteConstant.DEFAULT_TIMEOUT_MS));
            return (AskResponse) ask.toCompletableFuture().get(RemoteConstant.DEFAULT_TIMEOUT_MS, TimeUnit.MILLISECONDS);
        } catch (Exception e) {
            throw new AkkaCloudException(e);
        }
    }

    public static <T> T remoteAsk(ActorSelection remote, Object msg) {
        try {
            CompletionStage<T> ask = (CompletionStage<T>) Patterns.ask(remote, msg, Duration.ofMillis(RemoteConstant.DEFAULT_TIMEOUT_MS));
            return SerializerUtils.deSerialized((byte[]) ask.toCompletableFuture().get(RemoteConstant.DEFAULT_TIMEOUT_MS, TimeUnit.MILLISECONDS));
        } catch (Exception e) {
            throw new AkkaCloudException(e);
        }
    }

    public static <T> T remoteAskByRouter(ActorRef remote, Object msg) {
        try {
            CompletionStage<T> ask = (CompletionStage<T>) Patterns.ask(remote, msg, Duration.ofMillis(RemoteConstant.DEFAULT_TIMEOUT_MS));
            return SerializerUtils.deSerialized((byte[]) ask.toCompletableFuture().get(RemoteConstant.DEFAULT_TIMEOUT_MS, TimeUnit.MILLISECONDS));
        } catch (Exception e) {
            throw new AkkaCloudException(e);
        }
    }

    public static void remoteAskByRouterIgnore(ActorRef remote, Object msg) {
        Patterns.ask(remote, msg, Duration.ofMillis(RemoteConstant.DEFAULT_TIMEOUT_MS));
    }

    public static void remoteAskIgnore(ActorSelection remote, Object msg) {
        try {
            Patterns.ask(remote, msg, Duration.ofMillis(RemoteConstant.DEFAULT_TIMEOUT_MS));
        } catch (Exception e) {
            throw new AkkaCloudException(e);
        }
    }

    /**
     * 多次尝试出来的产物
     */
    public static <T> T handleWhenAsk(ActorSelection remote, Object msg, Supplier<T> executor) throws Exception {
        return Patterns.ask(remote, msg, Duration.ofMillis(RemoteConstant.DEFAULT_TIMEOUT_MS))
                .thenApply(ignore -> executor.get()).toCompletableFuture().get();
    }

    public static void handleWhenAskIgnore(ActorSelection remote, Object msg, Meaningless executor) {
        Patterns.ask(remote, msg, Duration.ofMillis(RemoteConstant.DEFAULT_TIMEOUT_MS)).whenComplete((response, throwable) -> {
            try {
                executor.m();
            } catch (Exception e) {
                throw new AkkaCloudException(e);
            }
        });
    }

    /**
     * create cluster router
     * select:
     * <p>0.RoundRobinRoutingLogic,1.RandomRoutingLogic,2.BroadcastRoutingLogic,
     * 3.ScatterGatherFirstCompletedRoutingLogic,4.ConsistentHashingRoutingLogic.
     * </p>
     */
    public static ActorRef createrRouter(ActorContext actorContext, String actorName, List<String> roleList, int totalInstances, int select) throws Exception{
        Iterable<String> routeesPaths = Lists.newArrayList("/user/" + actorName);
        Set<String> useRoles = new HashSet<>(roleList);
        switch (select) {
            case 0:
                return actorContext
                        .actorOf(
                                new ClusterRouterGroup(
                                        new RoundRobinGroup(routeesPaths),
                                        new ClusterRouterGroupSettings(
                                                totalInstances, routeesPaths, true, useRoles))
                                        .props(), actorName + "Router");
            case 1:
                return actorContext
                        .actorOf(
                                new ClusterRouterGroup(
                                        new RandomGroup(routeesPaths),
                                        new ClusterRouterGroupSettings(
                                                totalInstances, routeesPaths, true, useRoles))
                                        .props(), actorName);
            case 2:
                return actorContext
                        .actorOf(
                                new ClusterRouterGroup(
                                        new BroadcastGroup(routeesPaths),
                                        new ClusterRouterGroupSettings(
                                                totalInstances, routeesPaths, true, useRoles))
                                        .props(), actorName);
            case 3:
                return actorContext
                        .actorOf(
                                new ClusterRouterGroup(
                                        new ConsistentHashingGroup(routeesPaths),
                                        new ClusterRouterGroupSettings(
                                                totalInstances, routeesPaths, true, useRoles))
                                        .props(), actorName);
            default:
                throw new AkkaCloudException("can not recognized the identity.");
        }


    }
}
